package io.kiki.sba.registry.server.metadata.provide.data;

import io.kiki.sba.registry.common.model.console.PersistenceData;
import io.kiki.sba.registry.common.model.console.PersistenceDataBuilder;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.store.api.DBResponse;
import io.kiki.sba.registry.store.api.meta.ProvideDataRepository;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class DefaultProvideDataService implements ProvideDataService {

    private static final Logger logger = LoggerFactory.getLogger(DefaultProvideDataService.class);
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final ProvideDataRefresh provideDataRefresh = new ProvideDataRefresh();
    @Autowired
    MetadataLeaderService metadataLeaderService;
    private Map<String, PersistenceData> stringPersistenceDataMap = new ConcurrentHashMap<>();
    @Autowired
    private ProvideDataRepository provideDataRepository;

    @PostConstruct
    public void init() {
        ConcurrentUtils.createDaemonThread("provideData_refresh", provideDataRefresh).start();
        metadataLeaderService.registerListener(this);
    }

    private void provideDataRefresh() {
        if (!metadataLeaderService.amILeader()) {
            return;
        }
        Map<String, PersistenceData> stringPersistenceDataMap = provideDataRepository.getAll();

        logger.info("refresh provide data, old size: {}, new size: {}", this.stringPersistenceDataMap.size(), stringPersistenceDataMap.size());
        lock.writeLock().lock();
        try {
            this.stringPersistenceDataMap = stringPersistenceDataMap;
        } catch (Throwable t) {
            logger.error("refresh provide data error.", t);
        } finally {
            lock.writeLock().unlock();
        }
    }

    @Override
    public void becomeLeader() {
        provideDataRefresh.wakeup();
    }

    @Override
    public void loseLeader() {
    }


    @Override
    public boolean saveProvideData(PersistenceData persistenceData) {

        long expectVersion = getExpectVersion(PersistenceDataBuilder.getDataInfoId(persistenceData));
        return saveProvideData(persistenceData, expectVersion);
    }


    @Override
    public boolean saveProvideData(PersistenceData persistenceData, long expectVersion) {
        if (persistenceData.getVersion() <= expectVersion) {
            logger.error("save provide data: {} fail. new version: {} is smaller than old version: {}", persistenceData, persistenceData.getVersion(), expectVersion);
            return false;
        }

        // save with cas
        boolean success = provideDataRepository.put(persistenceData, expectVersion);

        if (success) {
            lock.writeLock().lock();
            try {
                // update local cache
                stringPersistenceDataMap.put(PersistenceDataBuilder.getDataInfoId(persistenceData), persistenceData);
            } catch (Throwable t) {
                logger.error("save provide data: {} error.", persistenceData, t);
                return false;
            } finally {
                lock.writeLock().unlock();
            }
        }
        return success;
    }

    private long getExpectVersion(String key) {
        long expectVersion = 0;
        DBResponse<PersistenceData> response = queryProvideData(key);

        if (response.getEntity() != null) {
            expectVersion = response.getEntity().getVersion();
        }
        return expectVersion;
    }

    @Override
    public DBResponse queryProvideData(String key) {
        PersistenceData persistenceData = null;
        lock.readLock().lock();
        try {
            persistenceData = stringPersistenceDataMap.get(key);
        } catch (Throwable t) {
            logger.error("query provide data: {} error.", key, t);
        } finally {
            lock.readLock().unlock();
        }

        if (persistenceData == null && !stringPersistenceDataMap.containsKey(key)) {
            return DBResponse.notfound().build();
        }
        if (persistenceData == null) {
            // not expect to visit this code;
            // if happen, query from persistent and print error log;
            persistenceData = provideDataRepository.get(key);
            logger.error("dataKey: {} not exist in cache, response from persistent: {}", key, persistenceData);
        }
        return persistenceData == null ? DBResponse.notfound().build() : DBResponse.ok(persistenceData).build();
    }

    @Override
    public boolean removeProvideData(String key) {
        long expectVersion = getExpectVersion(key);

        boolean success = provideDataRepository.remove(key, expectVersion);

        if (success) {
            lock.writeLock().lock();
            try {
                stringPersistenceDataMap.remove(key);
            } catch (Throwable t) {
                logger.error("remove provide data: {} error.", key, t);
                return false;
            } finally {
                lock.writeLock().unlock();
            }
        }
        return success;
    }

    private final class ProvideDataRefresh extends WakeUpLoopExecuteTask {

        @Override
        public void _execute_() {
            provideDataRefresh();
        }

        @Override
        public int getWaitingMillis() {
            return 3000;
        }
    }
}
